package io.github.bigbird0101.datatransfer.exchanger.impl;

import integration.data.task.api.channel.Channel;
import integration.data.task.api.transformation.impl.TransformerExecution;
import integration.modle.exception.IntegrationValidException;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.Communication;
import io.github.bigbird0101.datatransfer.collector.TaskPluginCollector;
import io.github.bigbird0101.datatransfer.exchanger.TransformerExchanger;
import io.github.bigbird0101.datatransfer.record.Record;
import io.github.bigbird0101.datatransfer.record.RecordReceiver;
import io.github.bigbird0101.datatransfer.record.RecordSender;
import io.github.bigbird0101.datatransfer.record.impl.DefaultRecord;
import io.github.bigbird0101.datatransfer.record.impl.TerminateRecord;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class RecordExchanger extends TransformerExchanger implements RecordSender, RecordReceiver {
    @Getter
    @Setter
    private Channel channel;

    private volatile boolean shutdown = false;

    public RecordExchanger() {
    }

    @SuppressWarnings("unchecked")
    public RecordExchanger(final String taskId, final Channel channel,
                           final Communication communication,
                           List<TransformerExecution> transformerExecs,
                           final TaskPluginCollector pluginCollector) {
        super(taskId, communication, transformerExecs, pluginCollector);
        assert channel != null;
        this.channel = channel;
    }

    public RecordExchanger(final String taskId, final Channel channel, final TaskPluginCollector pluginCollector) {
        super(taskId, null, null, pluginCollector);
        assert channel != null;
        this.channel = channel;
    }


    @Override
    public Record createRecord() {
        return new DefaultRecord();
    }

    @Override
    public void sendToWriter(Record record) {
        if (shutdown) {
            LogUtils.warn(log, "任务收到了shutdown指令");
//            return;
            throw new IntegrationValidException("shutdown");
        }
        record = doTransformer(record);
        if (record == null) {
            return;
        }
        this.channel.push(record);
        doStat();
    }

    @Override
    public void flush() {
        // do nothing
    }

    @Override
    public void terminate() {
//        if (shutdown) {
//            log.warn("任务已经收到了shutdown指令,写一个数据结束标记");
//            throw new IntegrationValidException("任务已经收到了shutdown指令");
//        }
        log.debug("任务收到了shutdown指令,写一个数据结束标记");
        this.channel.pushTerminate(TerminateRecord.get());
        doStat();
    }

    @Override
    public Record getFromReader() {
        Record record = this.channel.pull();
        return (record instanceof TerminateRecord ? null : record);
    }

    @Override
    public void shutdown() {
        shutdown = true;
        this.channel.close();
    }
}
